package com.ruoyi.iot.service.impl;

import com.ruoyi.common.config.RuoYiConfig;
import com.ruoyi.common.exception.ServiceException;
import com.ruoyi.common.utils.StringUtils;
import com.ruoyi.iot.domain.IotQueryResult;
import com.ruoyi.iot.domain.entity.DeviceAlarmWindow;
import com.ruoyi.iot.domain.entity.DeviceElectricalWindow;
import com.ruoyi.iot.domain.entity.DeviceFaultRecord;
import com.ruoyi.iot.domain.entity.DeviceFaultWindow;
import com.ruoyi.iot.domain.entity.DeviceTemperatureWindow;
import com.ruoyi.iot.mapper.DeviceAlarmWindowMapper;
import com.ruoyi.iot.mapper.DeviceElectricalWindowMapper;
import com.ruoyi.iot.mapper.DeviceFaultRecordMapper;
import com.ruoyi.iot.mapper.DeviceFaultWindowMapper;
import com.ruoyi.iot.mapper.DeviceTemperatureWindowMapper;
import com.ruoyi.iot.service.IotMetricExtractService;
import com.ruoyi.iot.service.IotQueryService;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.time.format.DateTimeFormatterBuilder;
import java.time.temporal.ChronoField;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.List;
import java.util.Locale;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

/**
 * IoTDB 指标抽取服务实现
 */
@Service
public class IotMetricExtractServiceImpl implements IotMetricExtractService {

    private static final Logger log = LoggerFactory.getLogger(IotMetricExtractServiceImpl.class);

    private static final ZoneId ZONE_ID = ZoneId.systemDefault();
    private static final DateTimeFormatter TIME_LITERAL_FORMATTER =
            DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS");
    private static final DateTimeFormatter FLEXIBLE_TIME_PARSER =
            new DateTimeFormatterBuilder()
                    .appendPattern("yyyy-MM-dd HH:mm:ss")
                    .optionalStart()
                    .appendFraction(ChronoField.MILLI_OF_SECOND, 1, 6, true)
                    .optionalEnd()
                    .toFormatter();
    private static final int TEMPERATURE_WINDOW_MINUTES = 10;
    private static final int ELECTRICAL_WINDOW_MINUTES = 10;
    private static final int ALARM_WINDOW_MINUTES = 5;
    private static final int FAULT_WINDOW_MINUTES = 30;
    private static final int DEFAULT_FETCH_WINDOW_COUNT = 12;
    private static final int FAULT_RECORD_LOOKBACK_MINUTES = ALARM_WINDOW_MINUTES * DEFAULT_FETCH_WINDOW_COUNT;

    private static final String TEMPERATURE_SQL_TEMPLATE_SJN =
            "SELECT device_id, date_bin(%1$dm, time) AS bucket_start, COUNT(*) AS sample_count, "
                    + "AVG(astatortemp1) AS avg_astatortemp1, AVG(bstatortemp1) AS avg_bstatortemp1, "
                    + "AVG(cstatortemp1) AS avg_cstatortemp1, AVG(astatortemp2) AS avg_astatortemp2, "
                    + "AVG(bstatortemp2) AS avg_bstatortemp2, AVG(cstatortemp2) AS avg_cstatortemp2, "
                    + "AVG(shaftendtemp) AS avg_shaftendtemp, AVG(nonshaftendtemp) AS avg_nonshaftendtemp, "
                    + "AVG(waterpumpbearingtemp) AS avg_waterpumpbearingtemp, "
                    + "MAX(astatortemp1) AS max_astatortemp1, MAX(bstatortemp1) AS max_bstatortemp1, "
                    + "MAX(cstatortemp1) AS max_cstatortemp1, MAX(astatortemp2) AS max_astatortemp2, "
                    + "MAX(bstatortemp2) AS max_bstatortemp2, MAX(cstatortemp2) AS max_cstatortemp2, "
                    + "MAX(shaftendtemp) AS max_shaftendtemp, MAX(nonshaftendtemp) AS max_nonshaftendtemp, "
                    + "MAX(waterpumpbearingtemp) AS max_waterpumpbearingtemp "
                    + "FROM product_1714885465641558018_data "
                    + "WHERE device_id = '%2$s' AND time >= %3$s AND time < %4$s "
                    + "GROUP BY device_id, date_bin(%1$dm, time) ORDER BY device_id, bucket_start";


    private static final String TEMPERATURE_SQL_TEMPLATE_JHT =
            "SELECT device_id, date_bin(%1$dm, time) AS bucket_start, COUNT(*) AS sample_count, "
                    + "AVG(astatortemp1) AS avg_astatortemp1, AVG(bstatortemp1) AS avg_bstatortemp1, "
                    + "AVG(cstatortemp1) AS avg_cstatortemp1, AVG(astatortemp2) AS avg_astatortemp2, "
                    + "AVG(bstatortemp2) AS avg_bstatortemp2, AVG(cstatortemp2) AS avg_cstatortemp2, "
                    + "AVG(shaftendtemp) AS avg_shaftendtemp, AVG(nonshaftendtemp) AS avg_nonshaftendtemp, "
                    + "MAX(astatortemp1) AS max_astatortemp1, MAX(bstatortemp1) AS max_bstatortemp1, "
                    + "MAX(cstatortemp1) AS max_cstatortemp1, MAX(astatortemp2) AS max_astatortemp2, "
                    + "MAX(bstatortemp2) AS max_bstatortemp2, MAX(cstatortemp2) AS max_cstatortemp2, "
                    + "MAX(shaftendtemp) AS max_shaftendtemp, MAX(nonshaftendtemp) AS max_nonshaftendtemp "

                    + "FROM product_1714885465641558018_data "
                    + "WHERE device_id = '%2$s' AND time >= %3$s AND time < %4$s "
                    + "GROUP BY device_id, date_bin(%1$dm, time) ORDER BY device_id, bucket_start";


    /*-------------------------------------------------------------------------------------------------------*/


    private static final String ELECTRICAL_SQL_TEMPLATE =
            "SELECT device_id, date_bin(%1$dm, time) AS bucket_start, COUNT(*) AS sample_count, "
                    + "AVG(ia) AS avg_ia, AVG(ib) AS avg_ib, AVG(ic) AS avg_ic, "
                    + "MAX(ia) AS max_ia, MAX(ib) AS max_ib, MAX(ic) AS max_ic, "
                    + "AVG(uab) AS avg_uab, AVG(ubc) AS avg_ubc, AVG(uca) AS avg_uca, "
                    + "AVG(powerfactor) AS avg_pf, SUM(p) * %1$d / 60.0 AS energy_kwh, "
                    + "SUM(q) * %1$d / 60.0 AS reactive_energy, "
                    + "CASE WHEN COUNT(*) > 0 THEN SUM(CASE WHEN quality = 0 THEN 1 ELSE 0 END) / COUNT(*) ELSE 0.0 END AS quality_rate "
                    + "FROM product_1964935588177780737_data "
                    + "WHERE device_id = '%2$s' AND time >= %3$s AND time < %4$s "
                    + "GROUP BY device_id, date_bin(%1$dm, time) ORDER BY device_id, bucket_start";

    private static final String ALARM_SQL_TEMPLATE_SJN =
            "SELECT device_id, date_bin(%1$dm, time) AS bucket_start, "
                    + "SUM(isquickstrikeaction) AS quickstrike_sum, "
                    + "SUM(isovercurrentprotectionaction) AS overcurrent_sum, "
                    + "SUM(isthermalprotectaction) AS thermal_sum, "
                    + "SUM(isoverpressureaction) AS overpressure_sum, "
                    + "SUM(isheatprotectedwarn) AS heat_warn_sum, "
                    + "SUM(iszerosequenceovercurrentalarm) AS zero_alarm_sum, "
                    + "SUM(isloadwarn) AS load_warn_sum, "
                    + "SUM(istotalalarms) AS total_alarm_cnt, "
                    + "SUM(istotalaccidents) AS total_fault_cnt "
                    + "FROM product_1714885465641558018_data "
                    + "WHERE device_id = '%2$s' AND time >= %3$s AND time < %4$s "
                    + "GROUP BY device_id, date_bin(%1$dm, time) ORDER BY device_id, bucket_start";

    private static final String JHT_FAULT_INDICATOR_SUM =
            "COALESCE(issettinggroupinvalid,0) + COALESCE(issettingcheckfault,0) + "
                    + "COALESCE(isiofault,0) + COALESCE(issamplingfault,0) + COALESCE(isprogramfault,0) + "
                    + "COALESCE(ishsbbusfault,0) + COALESCE(isinternalcommfault,0) + COALESCE(istimesyncfault,0) + "
                    + "COALESCE(isbmimodulefault,0) + COALESCE(isvtfault,0) + COALESCE(iscontrolcircuitfault,0) + "
                    + "COALESCE(istripfail,0) + COALESCE(ispositionmismatch,0)";

    private static final String JHT_ALARM_INDICATOR_SUM =
            "COALESCE(ismonitormodulealarm,0) + COALESCE(isfiberpoweralarm,0) + COALESCE(isovertemperaturealarm,0) + "
                    + "COALESCE(isoverloadalarm,0) + COALESCE(isovervoltagealarm,0) + COALESCE(iszeroseqvoltagealarm,0) + "
                    + "COALESCE(isnonelectrical1alarm,0) + COALESCE(isnonelectrical2alarm,0) + "
                    + "COALESCE(isnonelectrical3alarm,0) + COALESCE(iszeroseqovercurrentalarm,0) + "
                    + "COALESCE(iscontrolpowerfail,0) + COALESCE(isovertempwarn,0)";

    private static final String ALARM_SQL_TEMPLATE_JHT =
            "SELECT device_id, date_bin(%1$dm, time) AS bucket_start, "
                    + "SUM(0) AS quickstrike_sum, "
                    + "SUM(0) AS overcurrent_sum, "
                    + "SUM(0) AS thermal_sum, "
                    + "SUM(0) AS overpressure_sum, "
                    + "SUM(COALESCE(isovertempwarn, 0)) AS heat_warn_sum, "
                    + "SUM(COALESCE(iszeroseqovercurrentalarm, 0)) AS zero_alarm_sum, "
                    + "SUM(COALESCE(isoverloadalarm, 0)) AS load_warn_sum, "
                    + "SUM(" + JHT_ALARM_INDICATOR_SUM + ") AS total_alarm_cnt, "
                    + "SUM(" + JHT_FAULT_INDICATOR_SUM + ") AS total_fault_cnt, "
                    + "SUM(COALESCE(ismonitormodulealarm, 0)) AS monitor_module_alarm_sum, "
                    + "SUM(COALESCE(isfiberpoweralarm, 0)) AS fiber_power_alarm_sum, "
                    + "SUM(COALESCE(isovertemperaturealarm, 0)) AS over_temperature_alarm_sum, "
                    + "SUM(COALESCE(isovervoltagealarm, 0)) AS over_voltage_alarm_sum, "
                    + "SUM(COALESCE(iszeroseqvoltagealarm, 0)) AS zero_seq_voltage_alarm_sum, "
                    + "SUM(COALESCE(isnonelectrical1alarm, 0)) AS non_electrical1_alarm_sum, "
                    + "SUM(COALESCE(isnonelectrical2alarm, 0)) AS non_electrical2_alarm_sum, "
                    + "SUM(COALESCE(isnonelectrical3alarm, 0)) AS non_electrical3_alarm_sum, "
                    + "SUM(COALESCE(iscontrolpowerfail, 0)) AS control_power_fail_sum "
                    + "FROM product_1964935588177780737_data "
                    + "WHERE device_id = '%2$s' AND time >= %3$s AND time < %4$s "
                    + "GROUP BY device_id, date_bin(%1$dm, time) ORDER BY device_id, bucket_start";

    private static final String FAULT_SQL_TEMPLATE_SJN =
            "SELECT device_id, date_bin(%1$dm, time) AS bucket_start, "
                    + "SUM(CASE WHEN istotalaccidents > 0 THEN 1 ELSE 0 END) AS fault_event_cnt, "
                    + "SUM(istotalaccidents) AS fault_total_cnt, "
                    + "SUM(istotalalarms) AS alarm_total_cnt "
                    + "FROM product_1714885465641558018_data "
                    + "WHERE device_id = '%2$s' AND time >= %3$s AND time < %4$s "
                    + "GROUP BY device_id, date_bin(%1$dm, time) ORDER BY device_id, bucket_start";

    private static final String FAULT_SQL_TEMPLATE_JHT =
            "SELECT device_id, date_bin(%1$dm, time) AS bucket_start, "
                    + "SUM(" + JHT_FAULT_INDICATOR_SUM + ") AS fault_event_cnt, "
                    + "SUM(" + JHT_FAULT_INDICATOR_SUM + ") AS fault_total_cnt, "
                    + "SUM(" + JHT_ALARM_INDICATOR_SUM + ") AS alarm_total_cnt, "
                    + "SUM(COALESCE(issettinggroupinvalid,0)) AS setting_group_invalid_cnt, "
                    + "SUM(COALESCE(issettingcheckfault,0)) AS setting_check_fault_cnt, "
                    + "SUM(COALESCE(isiofault,0)) AS io_fault_cnt, "
                    + "SUM(COALESCE(issamplingfault,0)) AS sampling_fault_cnt, "
                    + "SUM(COALESCE(isprogramfault,0)) AS program_fault_cnt, "
                    + "SUM(COALESCE(ismonitormodulealarm,0)) AS monitor_module_alarm_cnt, "
                    + "SUM(COALESCE(isfiberpoweralarm,0)) AS fiber_power_alarm_cnt, "
                    + "SUM(COALESCE(ishsbbusfault,0)) AS hsb_bus_fault_cnt, "
                    + "SUM(COALESCE(isinternalcommfault,0)) AS internal_comm_fault_cnt, "
                    + "SUM(COALESCE(istimesyncfault,0)) AS time_sync_fault_cnt, "
                    + "SUM(COALESCE(isbmimodulefault,0)) AS bmi_module_fault_cnt, "
                    + "SUM(COALESCE(isvtfault,0)) AS tv_fault_cnt, "
                    + "SUM(COALESCE(iscontrolcircuitfault,0)) AS control_circuit_fault_cnt, "
                    + "SUM(COALESCE(istripfail,0)) AS trip_fail_cnt, "
                    + "SUM(COALESCE(ispositionmismatch,0)) AS position_mismatch_cnt, "
                    + "SUM(COALESCE(isovertemperaturealarm,0)) AS over_temperature_alarm_cnt, "
                    + "SUM(COALESCE(isoverloadalarm,0)) AS over_load_alarm_cnt, "
                    + "SUM(COALESCE(isovervoltagealarm,0)) AS over_voltage_alarm_cnt, "
                    + "SUM(COALESCE(iszeroseqvoltagealarm,0)) AS zero_seq_voltage_alarm_cnt, "
                    + "SUM(COALESCE(isnonelectrical1alarm,0)) AS non_electrical1_alarm_cnt, "
                    + "SUM(COALESCE(isnonelectrical2alarm,0)) AS non_electrical2_alarm_cnt, "
                    + "SUM(COALESCE(isnonelectrical3alarm,0)) AS non_electrical3_alarm_cnt, "
                    + "SUM(COALESCE(iszeroseqovercurrentalarm,0)) AS zero_seq_overcurrent_alarm_cnt, "
                    + "SUM(COALESCE(iscontrolpowerfail,0)) AS control_power_fail_cnt, "
                    + "SUM(COALESCE(isovertempwarn,0)) AS over_temp_warn_cnt "
                    + "FROM product_1964935588177780737_data "
                    + "WHERE device_id = '%2$s' AND time >= %3$s AND time < %4$s "
                    + "GROUP BY device_id, date_bin(%1$dm, time) ORDER BY device_id, bucket_start";

    private static final String FAULT_RECORD_SQL_TEMPLATE =
            "SELECT device_id, time AS event_time, "
                    + "isquickstrikeaction AS quickstrike_action, "
                    + "isovercurrentprotectionaction AS overcurrent_protection_action, "
                    + "isthermalprotectaction AS thermal_protect_action, "
                    + "isoverpressureaction AS overpressure_action, "
                    + "isheatprotectedwarn AS heat_protected_warn, "
                    + "iszerosequenceovercurrentalarm AS zero_sequence_overcurrent_alarm, "
                    + "isloadwarn AS load_warn, "
                    + "istotalalarms AS total_alarm_count, "
                    + "istotalaccidents AS total_fault_count "
                    + "FROM product_1714885465641558018_data "
                    + "WHERE device_id = '%1$s' AND time > %2$s AND time <= %3$s "
                    + "ORDER BY time ASC";

    private final IotQueryService queryService;
    private final DeviceTemperatureWindowMapper temperatureWindowMapper;
    private final DeviceElectricalWindowMapper electricalWindowMapper;
    private final DeviceAlarmWindowMapper alarmWindowMapper;
    private final DeviceFaultWindowMapper faultWindowMapper;
    private final DeviceFaultRecordMapper faultRecordMapper;

    public IotMetricExtractServiceImpl(IotQueryService queryService,
                                       DeviceTemperatureWindowMapper temperatureWindowMapper,
                                       DeviceElectricalWindowMapper electricalWindowMapper,
                                       DeviceAlarmWindowMapper alarmWindowMapper,
                                       DeviceFaultWindowMapper faultWindowMapper,
                                       DeviceFaultRecordMapper faultRecordMapper) {
        this.queryService = queryService;
        this.temperatureWindowMapper = temperatureWindowMapper;
        this.electricalWindowMapper = electricalWindowMapper;
        this.alarmWindowMapper = alarmWindowMapper;
        this.faultWindowMapper = faultWindowMapper;
        this.faultRecordMapper = faultRecordMapper;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void syncAllWindows(String deviceId) {
        // 此方法保留兼容性，实际任务由批量任务调用
        syncTemperatureWindows(deviceId);
        syncElectricalWindows(deviceId);
        syncAlarmWindows(deviceId);
        syncFaultWindows(deviceId);
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void syncTemperatureWindows(String deviceId) {
        validateDeviceId(deviceId);
        LocalDateTime endExclusive = alignToWindow(LocalDateTime.now(ZONE_ID), TEMPERATURE_WINDOW_MINUTES);
        LocalDateTime lastEnd = temperatureWindowMapper.selectLatestWindowEnd(deviceId, TEMPERATURE_WINDOW_MINUTES);
        LocalDateTime startInclusive = resolveStart(lastEnd, endExclusive, TEMPERATURE_WINDOW_MINUTES);
        if (startInclusive == null || !startInclusive.isBefore(endExclusive)) {
            log.debug("温度窗口无待抽取数据 deviceId={}, lastEnd={}, endExclusive={}", deviceId, lastEnd, endExclusive);
            return;
        }
        String sql = "";

        String area = RuoYiConfig.getArea();
        if (StringUtils.equals(area, "jht")) {
            sql = String.format(Locale.ROOT, TEMPERATURE_SQL_TEMPLATE_JHT,
                    TEMPERATURE_WINDOW_MINUTES, escapeDeviceId(deviceId),
                    formatTimeLiteral(startInclusive), formatTimeLiteral(endExclusive));
        } else {
            sql = String.format(Locale.ROOT, TEMPERATURE_SQL_TEMPLATE_SJN,
                    TEMPERATURE_WINDOW_MINUTES, escapeDeviceId(deviceId),
                    formatTimeLiteral(startInclusive), formatTimeLiteral(endExclusive));
        }


        log.info("执行温度窗口抽取SQL: {}", sql);
        IotQueryResult result = queryService.executeQuery(sql);

        List<DeviceTemperatureWindow> entities = new ArrayList<>();
        LocalDateTime now = LocalDateTime.now(ZONE_ID);
        for (Map<String, Object> row : result.getRows()) {
            LocalDateTime bucketStart = toLocalDateTime(row.get("bucket_start"));
            if (bucketStart == null || !bucketStart.isBefore(endExclusive)) {
                continue;
            }
            DeviceTemperatureWindow entity = new DeviceTemperatureWindow();
            entity.setDeviceId(deviceId);
            entity.setGranularityMinutes(TEMPERATURE_WINDOW_MINUTES);
            entity.setWindowStart(bucketStart);
            entity.setWindowEnd(bucketStart.plusMinutes(TEMPERATURE_WINDOW_MINUTES));
            entity.setSampleCount(toInteger(row.get("sample_count")));
            entity.setAvgAstatortemp1(toBigDecimal(row.get("avg_astatortemp1")));
            entity.setAvgBstatortemp1(toBigDecimal(row.get("avg_bstatortemp1")));
            entity.setAvgCstatortemp1(toBigDecimal(row.get("avg_cstatortemp1")));
            entity.setAvgAstatortemp2(toBigDecimal(row.get("avg_astatortemp2")));
            entity.setAvgBstatortemp2(toBigDecimal(row.get("avg_bstatortemp2")));
            entity.setAvgCstatortemp2(toBigDecimal(row.get("avg_cstatortemp2")));
            entity.setAvgShaftendtemp(toBigDecimal(row.get("avg_shaftendtemp")));
            entity.setAvgNonshaftendtemp(toBigDecimal(row.get("avg_nonshaftendtemp")));
            //entity.setAvgWaterpumpbearingtemp(toBigDecimal(row.get("avg_waterpumpbearingtemp")));
            entity.setMaxAstatortemp1(toBigDecimal(row.get("max_astatortemp1")));
            entity.setMaxBstatortemp1(toBigDecimal(row.get("max_bstatortemp1")));
            entity.setMaxCstatortemp1(toBigDecimal(row.get("max_cstatortemp1")));
            entity.setMaxAstatortemp2(toBigDecimal(row.get("max_astatortemp2")));
            entity.setMaxBstatortemp2(toBigDecimal(row.get("max_bstatortemp2")));
            entity.setMaxCstatortemp2(toBigDecimal(row.get("max_cstatortemp2")));
            entity.setMaxShaftendtemp(toBigDecimal(row.get("max_shaftendtemp")));
            entity.setMaxNonshaftendtemp(toBigDecimal(row.get("max_nonshaftendtemp")));
            // entity.setMaxWaterpumpbearingtemp(toBigDecimal(row.get("max_waterpumpbearingtemp")));
            entity.setCreatedAt(now);
            entity.setUpdatedAt(now);
            entities.add(entity);
        }
        if (!entities.isEmpty()) {
            temperatureWindowMapper.insertBatch(entities);
            log.info("同步温度窗口完成 deviceId={}, count={}", deviceId, entities.size());
        }
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void syncElectricalWindows(String deviceId) {
        validateDeviceId(deviceId);
        LocalDateTime endExclusive = alignToWindow(LocalDateTime.now(ZONE_ID), ELECTRICAL_WINDOW_MINUTES);
        LocalDateTime lastEnd = electricalWindowMapper.selectLatestWindowEnd(deviceId, ELECTRICAL_WINDOW_MINUTES);
        LocalDateTime startInclusive = resolveStart(lastEnd, endExclusive, ELECTRICAL_WINDOW_MINUTES);
        if (startInclusive == null || !startInclusive.isBefore(endExclusive)) {
            log.debug("电气窗口无待抽取数据 deviceId={}, lastEnd={}, endExclusive={}", deviceId, lastEnd, endExclusive);
            return;
        }
        String sql = String.format(Locale.ROOT, ELECTRICAL_SQL_TEMPLATE,
                ELECTRICAL_WINDOW_MINUTES, escapeDeviceId(deviceId),
                formatTimeLiteral(startInclusive), formatTimeLiteral(endExclusive));
        log.info("执行电气窗口抽取SQL: {}", sql);
        IotQueryResult result = queryService.executeQuery(sql);

        List<DeviceElectricalWindow> entities = new ArrayList<>();
        LocalDateTime now = LocalDateTime.now(ZONE_ID);
        for (Map<String, Object> row : result.getRows()) {
            LocalDateTime bucketStart = toLocalDateTime(row.get("bucket_start"));
            if (bucketStart == null || !bucketStart.isBefore(endExclusive)) {
                continue;
            }
            DeviceElectricalWindow entity = new DeviceElectricalWindow();
            entity.setDeviceId(deviceId);
            entity.setGranularityMinutes(ELECTRICAL_WINDOW_MINUTES);
            entity.setWindowStart(bucketStart);
            entity.setWindowEnd(bucketStart.plusMinutes(ELECTRICAL_WINDOW_MINUTES));
            entity.setSampleCount(toInteger(row.get("sample_count")));
            entity.setAvgIa(toBigDecimal(row.get("avg_ia")));
            entity.setAvgIb(toBigDecimal(row.get("avg_ib")));
            entity.setAvgIc(toBigDecimal(row.get("avg_ic")));
            entity.setMaxIa(toBigDecimal(row.get("max_ia")));
            entity.setMaxIb(toBigDecimal(row.get("max_ib")));
            entity.setMaxIc(toBigDecimal(row.get("max_ic")));
            entity.setAvgUab(toBigDecimal(row.get("avg_uab")));
            entity.setAvgUbc(toBigDecimal(row.get("avg_ubc")));
            entity.setAvgUca(toBigDecimal(row.get("avg_uca")));
            entity.setAvgPowerfactor(toBigDecimal(row.get("avg_pf")));
            entity.setEnergyKwh(toBigDecimal(row.get("energy_kwh")));
            entity.setReactiveEnergy(toBigDecimal(row.get("reactive_energy")));
            entity.setQualityRate(safeScale(toBigDecimal(row.get("quality_rate")), 6));
            entity.setCreatedAt(now);
            entity.setUpdatedAt(now);
            entities.add(entity);
        }
        if (!entities.isEmpty()) {
            electricalWindowMapper.insertBatch(entities);
            log.info("同步电气窗口完成 deviceId={}, count={}", deviceId, entities.size());
        }
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void syncAlarmWindows(String deviceId) {
        validateDeviceId(deviceId);
        LocalDateTime endExclusive = alignToWindow(LocalDateTime.now(ZONE_ID), ALARM_WINDOW_MINUTES);
        LocalDateTime lastEnd = alarmWindowMapper.selectLatestWindowEnd(deviceId, ALARM_WINDOW_MINUTES);
        LocalDateTime startInclusive = resolveStart(lastEnd, endExclusive, ALARM_WINDOW_MINUTES);
        if (startInclusive == null || !startInclusive.isBefore(endExclusive)) {
            log.debug("告警窗口无待抽取数据 deviceId={}, lastEnd={}, endExclusive={}", deviceId, lastEnd, endExclusive);
            return;
        }
        String area = RuoYiConfig.getArea();
        boolean isJht = StringUtils.equals(area, "jht");
        String alarmTemplate = isJht ? ALARM_SQL_TEMPLATE_JHT : ALARM_SQL_TEMPLATE_SJN;
        String sql = String.format(Locale.ROOT, alarmTemplate,
                ALARM_WINDOW_MINUTES, escapeDeviceId(deviceId),
                formatTimeLiteral(startInclusive), formatTimeLiteral(endExclusive));

        log.info("执行告警窗口抽取SQL: {}", sql);
        IotQueryResult result = queryService.executeQuery(sql);


        List<DeviceAlarmWindow> entities = new ArrayList<>();
        LocalDateTime now = LocalDateTime.now(ZONE_ID);
        for (Map<String, Object> row : result.getRows()) {
            LocalDateTime bucketStart = toLocalDateTime(row.get("bucket_start"));
            if (bucketStart == null || !bucketStart.isBefore(endExclusive)) {
                continue;
            }
            DeviceAlarmWindow entity = new DeviceAlarmWindow();
            entity.setDeviceId(deviceId);
            entity.setGranularityMinutes(ALARM_WINDOW_MINUTES);
            entity.setWindowStart(bucketStart);
            entity.setWindowEnd(bucketStart.plusMinutes(ALARM_WINDOW_MINUTES));
            entity.setQuickstrikeSum(toInteger(row.get("quickstrike_sum")));
            entity.setOvercurrentSum(toInteger(row.get("overcurrent_sum")));
            entity.setThermalSum(toInteger(row.get("thermal_sum")));
            entity.setOverpressureSum(toInteger(row.get("overpressure_sum")));
            entity.setHeatWarnSum(toInteger(row.get("heat_warn_sum")));
            entity.setZeroAlarmSum(toInteger(row.get("zero_alarm_sum")));
            entity.setLoadWarnSum(toInteger(row.get("load_warn_sum")));
            entity.setTotalAlarmCount(toInteger(row.get("total_alarm_cnt")));
            entity.setTotalFaultCount(toInteger(row.get("total_fault_cnt")));
            if (isJht) {
                entity.setMonitorModuleAlarmSum(toInteger(row.get("monitor_module_alarm_sum")));
                entity.setFiberPowerAlarmSum(toInteger(row.get("fiber_power_alarm_sum")));
                entity.setOverTemperatureAlarmSum(toInteger(row.get("over_temperature_alarm_sum")));
                entity.setOverVoltageAlarmSum(toInteger(row.get("over_voltage_alarm_sum")));
                entity.setZeroSeqVoltageAlarmSum(toInteger(row.get("zero_seq_voltage_alarm_sum")));
                entity.setNonElectrical1AlarmSum(toInteger(row.get("non_electrical1_alarm_sum")));
                entity.setNonElectrical2AlarmSum(toInteger(row.get("non_electrical2_alarm_sum")));
                entity.setNonElectrical3AlarmSum(toInteger(row.get("non_electrical3_alarm_sum")));
                entity.setControlPowerFailSum(toInteger(row.get("control_power_fail_sum")));
            }
            entity.setCreatedAt(now);
            entity.setUpdatedAt(now);
            entities.add(entity);
        }
        if (!entities.isEmpty()) {
            alarmWindowMapper.insertBatch(entities);
            log.info("同步告警窗口完成 deviceId={}, count={}", deviceId, entities.size());
        }
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void syncFaultWindows(String deviceId) {
        validateDeviceId(deviceId);
        LocalDateTime endExclusive = alignToWindow(LocalDateTime.now(ZONE_ID), FAULT_WINDOW_MINUTES);
        LocalDateTime lastEnd = faultWindowMapper.selectLatestWindowEnd(deviceId, FAULT_WINDOW_MINUTES);
        LocalDateTime startInclusive = resolveStart(lastEnd, endExclusive, FAULT_WINDOW_MINUTES);
        if (startInclusive == null || !startInclusive.isBefore(endExclusive)) {
            log.debug("故障窗口无待抽取数据 deviceId={}, lastEnd={}, endExclusive={}", deviceId, lastEnd, endExclusive);
            return;
        }
        String area = RuoYiConfig.getArea();
        boolean isJht = StringUtils.equals(area, "jht");
        String template = isJht ? FAULT_SQL_TEMPLATE_JHT : FAULT_SQL_TEMPLATE_SJN;
        String sql = String.format(Locale.ROOT, template,
                FAULT_WINDOW_MINUTES, escapeDeviceId(deviceId),
                formatTimeLiteral(startInclusive), formatTimeLiteral(endExclusive));
        log.info("执行故障窗口抽取SQL: {}", sql);
        IotQueryResult result = queryService.executeQuery(sql);

        List<DeviceFaultWindow> entities = new ArrayList<>();
        LocalDateTime now = LocalDateTime.now(ZONE_ID);
        for (Map<String, Object> row : result.getRows()) {
            LocalDateTime bucketStart = toLocalDateTime(row.get("bucket_start"));
            if (bucketStart == null || !bucketStart.isBefore(endExclusive)) {
                continue;
            }
            DeviceFaultWindow entity = new DeviceFaultWindow();
            entity.setDeviceId(deviceId);
            entity.setGranularityMinutes(FAULT_WINDOW_MINUTES);
            entity.setWindowStart(bucketStart);
            entity.setWindowEnd(bucketStart.plusMinutes(FAULT_WINDOW_MINUTES));
            entity.setFaultEventCount(toInteger(row.get("fault_event_cnt")));
            entity.setTotalFaultCount(toInteger(row.get("fault_total_cnt")));
            entity.setAlarmTotalCount(toInteger(row.get("alarm_total_cnt")));
            if (isJht) {
                entity.setSettingGroupInvalidCount(toInteger(row.get("setting_group_invalid_cnt")));
                entity.setSettingCheckFaultCount(toInteger(row.get("setting_check_fault_cnt")));
                entity.setIoFaultCount(toInteger(row.get("io_fault_cnt")));
                entity.setSamplingFaultCount(toInteger(row.get("sampling_fault_cnt")));
                entity.setProgramFaultCount(toInteger(row.get("program_fault_cnt")));
                entity.setMonitorModuleAlarmCount(toInteger(row.get("monitor_module_alarm_cnt")));
                entity.setFiberPowerAlarmCount(toInteger(row.get("fiber_power_alarm_cnt")));
                entity.setHsbBusFaultCount(toInteger(row.get("hsb_bus_fault_cnt")));
                entity.setInternalCommFaultCount(toInteger(row.get("internal_comm_fault_cnt")));
                entity.setTimeSyncFaultCount(toInteger(row.get("time_sync_fault_cnt")));
                entity.setBmiModuleFaultCount(toInteger(row.get("bmi_module_fault_cnt")));
                entity.setTvFaultCount(toInteger(row.get("tv_fault_cnt")));
                entity.setControlCircuitFaultCount(toInteger(row.get("control_circuit_fault_cnt")));
                entity.setTripFailCount(toInteger(row.get("trip_fail_cnt")));
                entity.setPositionMismatchCount(toInteger(row.get("position_mismatch_cnt")));
                entity.setOverTemperatureAlarmCount(toInteger(row.get("over_temperature_alarm_cnt")));
                entity.setOverLoadAlarmCount(toInteger(row.get("over_load_alarm_cnt")));
                entity.setOverVoltageAlarmCount(toInteger(row.get("over_voltage_alarm_cnt")));
                entity.setZeroSeqVoltageAlarmCount(toInteger(row.get("zero_seq_voltage_alarm_cnt")));
                entity.setNonElectrical1AlarmCount(toInteger(row.get("non_electrical1_alarm_cnt")));
                entity.setNonElectrical2AlarmCount(toInteger(row.get("non_electrical2_alarm_cnt")));
                entity.setNonElectrical3AlarmCount(toInteger(row.get("non_electrical3_alarm_cnt")));
                entity.setZeroSeqOvercurrentAlarmCount(toInteger(row.get("zero_seq_overcurrent_alarm_cnt")));
                entity.setControlPowerFailCount(toInteger(row.get("control_power_fail_cnt")));
                entity.setOverTempWarnCount(toInteger(row.get("over_temp_warn_cnt")));
            }
            entity.setCreatedAt(now);
            entity.setUpdatedAt(now);
            entities.add(entity);
        }
        if (!entities.isEmpty()) {
            faultWindowMapper.insertBatch(entities);
            log.info("同步故障窗口完成 deviceId={}, count={}", deviceId, entities.size());
        }
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void syncFaultRecords(String deviceId) {
        validateDeviceId(deviceId);
        LocalDateTime now = LocalDateTime.now(ZONE_ID);
        LocalDateTime endExclusive = now;
        LocalDateTime lastEvent = faultRecordMapper.selectLatestEventTime(deviceId);
        LocalDateTime startExclusive = resolveRecordStart(lastEvent, endExclusive);
        if (startExclusive == null || !startExclusive.isBefore(endExclusive)) {
            log.debug("故障记录无待抽取数据 deviceId={}, lastEvent={}, endExclusive={}", deviceId, lastEvent, endExclusive);
            return;
        }
        String sql = String.format(Locale.ROOT, FAULT_RECORD_SQL_TEMPLATE,
                escapeDeviceId(deviceId),
                formatTimeLiteral(startExclusive),
                formatTimeLiteral(endExclusive));
        IotQueryResult result = queryService.executeQuery(sql);
        List<Map<String, Object>> rows = result.getRows();
        if (rows.isEmpty()) {
            return;
        }
        List<DeviceFaultRecord> records = new ArrayList<>();
        LocalDateTime current = LocalDateTime.now(ZONE_ID);
        for (Map<String, Object> row : rows) {
            LocalDateTime eventTime = toExactLocalDateTime(row.get("event_time"));
            if (eventTime == null) {
                continue;
            }
            if (lastEvent != null && !eventTime.isAfter(lastEvent)) {
                continue;
            }
            Integer quick = toInteger(row.get("quickstrike_action"));
            Integer overcurrent = toInteger(row.get("overcurrent_protection_action"));
            Integer thermal = toInteger(row.get("thermal_protect_action"));
            Integer overpressure = toInteger(row.get("overpressure_action"));
            Integer heatWarn = toInteger(row.get("heat_protected_warn"));
            Integer zeroAlarm = toInteger(row.get("zero_sequence_overcurrent_alarm"));
            Integer loadWarn = toInteger(row.get("load_warn"));
            Integer totalAlarm = toInteger(row.get("total_alarm_count"));
            Integer totalFault = toInteger(row.get("total_fault_count"));
            if (isAllZero(quick, overcurrent, thermal, overpressure, heatWarn, zeroAlarm, loadWarn,
                    totalAlarm, totalFault)) {
                continue;
            }
            DeviceFaultRecord record = new DeviceFaultRecord();
            record.setDeviceId(deviceId);
            record.setEventTime(eventTime);
            record.setQuickStrikeAction(quick);
            record.setOvercurrentProtectionAction(overcurrent);
            record.setThermalProtectAction(thermal);
            record.setOverpressureAction(overpressure);
            record.setHeatProtectedWarn(heatWarn);
            record.setZeroSequenceOvercurrentAlarm(zeroAlarm);
            record.setLoadWarn(loadWarn);
            record.setTotalAlarmCount(totalAlarm);
            record.setTotalFaultCount(totalFault);
            record.setCreatedAt(current);
            record.setUpdatedAt(current);
            records.add(record);
        }
        if (!records.isEmpty()) {
            faultRecordMapper.insertBatch(records);
            log.info("同步故障明细完成 deviceId={}, count={}", deviceId, records.size());
        }
    }

    private void validateDeviceId(String deviceId) {
        if (StringUtils.isBlank(deviceId)) {
            throw new ServiceException("设备编号不能为空");
        }
    }

    private LocalDateTime resolveRecordStart(LocalDateTime lastEvent, LocalDateTime endExclusive) {
        if (endExclusive == null) {
            return null;
        }
        if (lastEvent != null) {
            if (!lastEvent.isBefore(endExclusive)) {
                return null;
            }
            return lastEvent;
        }
        return endExclusive.minusMinutes(FAULT_RECORD_LOOKBACK_MINUTES);
    }

    private LocalDateTime resolveStart(LocalDateTime lastWindowEnd, LocalDateTime alignEnd, int windowMinutes) {
        if (alignEnd == null) {
            return null;
        }
        if (lastWindowEnd != null) {
            if (!lastWindowEnd.isBefore(alignEnd)) {
                return null;
            }
            return lastWindowEnd;
        }
        return alignEnd.minusMinutes((long) windowMinutes * DEFAULT_FETCH_WINDOW_COUNT);
    }

    private LocalDateTime alignToWindow(LocalDateTime time, int windowMinutes) {
        if (time == null) {
            return null;
        }
        LocalDateTime truncated = time.truncatedTo(ChronoUnit.MINUTES);
        int minute = truncated.getMinute();
        int offset = minute % windowMinutes;
        return truncated.minusMinutes(offset);
    }

    private LocalDateTime toLocalDateTime(Object value) {
        if (value == null) {
            return null;
        }
        if (value instanceof LocalDateTime) {
            return ((LocalDateTime) value).truncatedTo(ChronoUnit.MINUTES);
        }
        if (value instanceof Number) {
            long millis = ((Number) value).longValue();
            return LocalDateTime.ofInstant(Instant.ofEpochMilli(millis), ZONE_ID)
                    .truncatedTo(ChronoUnit.MINUTES);
        }
        if (value instanceof CharSequence) {
            try {
                return LocalDateTime.parse(value.toString(), FLEXIBLE_TIME_PARSER)
                        .truncatedTo(ChronoUnit.MINUTES);
            } catch (Exception ex) {
                log.debug("无法解析时间值: {}", value, ex);
            }
        }
        return null;
    }

    private LocalDateTime toExactLocalDateTime(Object value) {
        if (value == null) {
            return null;
        }
        if (value instanceof LocalDateTime) {
            return (LocalDateTime) value;
        }
        if (value instanceof Number) {
            long millis = ((Number) value).longValue();
            return LocalDateTime.ofInstant(Instant.ofEpochMilli(millis), ZONE_ID);
        }
        if (value instanceof CharSequence) {
            try {
                return LocalDateTime.parse(value.toString(), FLEXIBLE_TIME_PARSER);
            } catch (Exception ex) {
                log.debug("无法解析时间: {}", value, ex);
            }
        }
        return null;
    }

    private Integer toInteger(Object value) {
        if (value == null) {
            return null;
        }
        if (value instanceof Integer) {
            return (Integer) value;
        }
        if (value instanceof Number) {
            return ((Number) value).intValue();
        }
        if (value instanceof CharSequence) {
            try {
                return Integer.parseInt(value.toString());
            } catch (NumberFormatException ex) {
                log.debug("无法解析整型值: {}", value);
            }
        }
        return null;
    }

    private BigDecimal toBigDecimal(Object value) {
        if (value == null) {
            return null;
        }
        if (value instanceof BigDecimal) {
            return (BigDecimal) value;
        }
        if (value instanceof Number) {
            if (value instanceof Double || value instanceof Float) {
                return BigDecimal.valueOf(((Number) value).doubleValue());
            }
            return BigDecimal.valueOf(((Number) value).longValue());
        }
        if (value instanceof CharSequence) {
            try {
                return new BigDecimal(value.toString());
            } catch (NumberFormatException ex) {
                log.debug("无法解析数值: {}", value);
            }
        }
        return null;
    }

    private BigDecimal safeScale(BigDecimal value, int scale) {
        if (value == null) {
            return null;
        }
        return value.setScale(scale, RoundingMode.HALF_UP);
    }

    private boolean isAllZero(Integer... values) {
        if (values == null || values.length == 0) {
            return true;
        }
        for (Integer value : values) {
            if (value != null && value != 0) {
                return false;
            }
        }
        return true;
    }

    private String escapeDeviceId(String deviceId) {
        return deviceId.replace("'", "''");
    }

    private String formatTimeLiteral(LocalDateTime time) {
        return TIME_LITERAL_FORMATTER.format(time);
    }
}
